package beautiful.butterfly.drds.schedule.task.schedule_manager;

import beautiful.butterfly.drds.schedule.ITask;
import beautiful.butterfly.drds.schedule.ScheduleUtil;
import beautiful.butterfly.drds.schedule.TaskItem;
import beautiful.butterfly.drds.schedule.cron_expression.CronExpression;
import beautiful.butterfly.drds.schedule.strategy.IStrategyTask;
import beautiful.butterfly.drds.schedule.strategy.factory.ManageFactory;
import beautiful.butterfly.drds.schedule.task.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;
import java.util.List;
import java.util.Timer;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;


/**
 * 1��������ȷ�������Ŀ�꣺	�����е������ظ�������©�ı����ٴ���
 * 2��һ��Managerֻ����һ���������͵�һ�鹤���̡߳�
 * 3����һ��JVM������ܴ��ڶ��������ͬ�������͵�Manager��Ҳ���ܴ��ڴ���ͬ�������͵�Manager��
 * 4���ڲ�ͬ��JVM������Դ��ڴ�����ͬ�����Manager
 * 5�����ȵ�Manager���Զ�̬���������Ӻ�ֹͣ
 * <p>
 * ��Ҫ��ְ��
 * 1����ʱ���е������������ĸ��µ�ǰ���ȷ�����������״̬
 * 2���������������Ļ�ȡ���з�������״̬�����¼�������ķ��䡣��ô����Ŀ���Ǳ��⼯������������ĵĵ������⡣
 * 3����ÿ���������ݴ�����Ϻ󣬼���Ƿ���������������������Լ��ѳֵ�������У�����У����ͷŸ���ش����������
 * <p>
 * ������
 * �����ǰ�������ڴ���ǰ�����ʱ��ʱ����Ҫ�����ǰ���У����ͷ��Ѿ��ѳֵ����񡣲�������������ı�����
 *
 * @author xuannan
 */
@SuppressWarnings({"rawtypes", "unchecked"})
public abstract class Manager implements IStrategyTask {
    private static transient Logger logger = LoggerFactory.getLogger(Manager.class);
    /**
     * �û���ʶ��ͬ�̵߳����
     */
    private static int nextSerialNumber = 0;

    /**
     * ��ǰ�߳�����
     */
    protected int currentSerialNumber = 0;
    /**
     * ��������������Ϣ
     */
    protected TaskType taskTypeInfo;
    /**
     * ��ǰ���ȷ������Ϣ
     */
    protected Server currenServer;
    /**
     * ���д�����
     */
    ITask task;

    /**
     * ���߳���������
     */
    public IProcessor processor;
    Statistics statistics = new Statistics();

    public boolean isPauseSchedule = true;
    String pauseMessage = "";
    /**
     * ��ǰ������������嵥
     * ArrayListʵ�ֲ���ͬ���ġ�����̲߳����޸ĸ��б������ConcurrentModificationException
     */
    protected List<TaskItem> currentTaskItemList = new CopyOnWriteArrayList<TaskItem>();
    /**
     * ���һ������װ�ص��������ʱ�䡣
     * ��ǰʵ��  - �ϴ�װ��ʱ��  > intervalReloadTaskItemList���������������������µ�����������
     */
    protected long lastReloadTaskItemListTime = 0;
    protected boolean isNeedReloadTaskItem = true;


    private String mBeanName;
    /**
     * ���������ĸ�����Ϣ�Ķ�ʱ��
     */
    private Timer heartBeatTimer;

    public IDataManager scheduleCenter;

    protected String startErrorInfo = null;

    protected boolean isStopSchedule = false;
    protected Lock registerLock = new ReentrantLock();

    /**
     * ��������Ϣ�Ƿ��ʼ���ɹ�
     */
    protected boolean isRuntimeInfoInitial = false;

    ManageFactory manageFactory;

    public Manager(ManageFactory manageFactory, String baseTaskType, String ownSign, IDataManager aScheduleCenter) throws Exception {
        this.manageFactory = manageFactory;
        this.currentSerialNumber = serialNumber();
        this.scheduleCenter = aScheduleCenter;
        this.taskTypeInfo = this.scheduleCenter.loadTaskTypeBaseInfo(baseTaskType);
        logger.info("create Manager for taskType:" + baseTaskType);
        //����Ѿ�����1���TASK,OWN_SIGN����ϡ�����һ��û�лserver����Ϊ����
        this.scheduleCenter.clearExpireTaskTypeRunningInfo(baseTaskType, ScheduleUtil.getLocalIP() + "�������OWN_SIGN��Ϣ", this.taskTypeInfo.getExpireOwnSignInterval());

        Object dealBean = manageFactory.getBean(this.taskTypeInfo.getDealBeanName());
        if (dealBean == null) {
            throw new Exception("SpringBean " + this.taskTypeInfo.getDealBeanName() + " ������");
        }
        if (dealBean instanceof ITask == false) {
            throw new Exception("SpringBean " + this.taskTypeInfo.getDealBeanName() + " û��ʵ�� ITask�ӿ�");
        }
        this.task = (ITask) dealBean;

        if (this.taskTypeInfo.getJudgeDeadInterval() < this.taskTypeInfo.getHeartBeatRate() * 5) {
            throw new Exception("�������ô������⣬������ʱ����������Ҫ���������̵߳�5������ǰ�������ݣ�JudgeDeadInterval = "
                    + this.taskTypeInfo.getJudgeDeadInterval()
                    + ",HeartBeatRate = " + this.taskTypeInfo.getHeartBeatRate());
        }
        this.currenServer = Server.createScheduleServer(this.scheduleCenter, baseTaskType, ownSign, this.taskTypeInfo.getThreadNumber());
        this.currenServer.setManagerFactoryUUID(this.manageFactory.getUuid());
        scheduleCenter.registerScheduleServer(this.currenServer);
        this.mBeanName = "pamirs:name=" + "schedule.ServerMananger." + this.currenServer.getUuid();
        this.heartBeatTimer = new Timer(this.currenServer.getTaskType() + "-" + this.currentSerialNumber + "-HeartBeat");
        this.heartBeatTimer.schedule(new HeartBeatTimerTask(this),
                new Date(System.currentTimeMillis() + 500),
                this.taskTypeInfo.getHeartBeatRate());
        initial();
    }

    /**
     * ���󴴽�ʱ��Ҫ���ĳ�ʼ������
     *
     * @throws Exception
     */
    public abstract void initial() throws Exception;

    public abstract void refreshServerInfo() throws Exception;

    public abstract void assignScheduleTask() throws Exception;

    public abstract List<TaskItem> getCurrentScheduleTaskItemList();

    public abstract int getTaskItemCount();

    public String getTaskType() {
        return this.currenServer.getTaskType();
    }

    public void initialTaskParameter(String strategyName, String taskParameter) {
        //û��ʵ�ֵķ�������Ҫ�Ĳ���ֱ�Ӵ����������ж�ȡ
    }

    private static synchronized int serialNumber() {
        return nextSerialNumber++;
    }

    public int getCurrentSerialNumber() {
        return this.currentSerialNumber;
    }

    /**
     * ����ڴ������е��Ѿ�ȡ�õ����ݺ��������,����̬����ʧ�ܣ����߷���ע�����ĵĵ�����Ϣ��ɾ��
     */
    public void clearMemoInfo() {
        try {
            // ����ڴ������е��Ѿ�ȡ�õ����ݺ��������,����̬����ʧ�ܣ����߷���ע�����ĵĵ�����Ϣ��ɾ��
            this.currentTaskItemList.clear();
            if (this.processor != null) {
                this.processor.clearAllHasFetchData();
            }
        } finally {
            //�����ڴ����������������Ҫ����װ��
            this.isNeedReloadTaskItem = true;
        }

    }

    public void rewriteScheduleInfo() throws Exception {
        registerLock.lock();
        try {
            if (this.isStopSchedule == true) {
                if (logger.isDebugEnabled()) {
                    logger.debug("�ⲿ������ֹ����,����ע����ȷ��񣬱��������������ݣ�" + currenServer.getUuid());
                }
                return;
            }
            //�ȷ���������Ϣ
            if (startErrorInfo == null) {
                this.currenServer.setDealInfoDesc(this.pauseMessage + ":" + this.statistics.getDealDescription());
            } else {
                this.currenServer.setDealInfoDesc(startErrorInfo);
            }
            if (this.scheduleCenter.refreshScheduleServer(this.currenServer) == false) {
                //������Ϣʧ�ܣ�����ڴ����ݺ�����ע��
                this.clearMemoInfo();
                this.scheduleCenter.registerScheduleServer(this.currenServer);
            }
        } finally {
            registerLock.unlock();
        }
    }


    /**
     * ��ʼ��ʱ�򣬼����һ��ִ��ʱ��
     *
     * @throws Exception
     */
    public void computerStart() throws Exception {
        //ֻ�е����ڿ�ִ�ж��к��ٿ�ʼ��������

        boolean isRunNow = false;
        if (this.taskTypeInfo.getPermitRunStartTime() == null) {
            isRunNow = true;
        } else {
            String tmpStr = this.taskTypeInfo.getPermitRunStartTime();
            if (tmpStr.toLowerCase().startsWith("startrun:")) {
                isRunNow = true;
                tmpStr = tmpStr.substring("startrun:".length());
            }
            CronExpression cexpStart = new CronExpression(tmpStr);
            Date current = new Date(this.scheduleCenter.getSystemTime());
            Date firstStartTime = cexpStart.getNextValidTimeAfter(current);
            this.heartBeatTimer.schedule(
                    new PauseOrResumeTask(this, this.heartBeatTimer,
                            PauseOrResumeTask.TYPE_RESUME, tmpStr),
                    firstStartTime);
            this.currenServer.setNextRunStartTime(ScheduleUtil.transferDataToString(firstStartTime));
            if (this.taskTypeInfo.getPermitRunEndTime() == null
                    || this.taskTypeInfo.getPermitRunEndTime().equals("-1")) {
                this.currenServer.setNextRunEndTime("�����ܻ�ȡ�����ݵ�ʱ��pause");
            } else {
                try {
                    String tmpEndStr = this.taskTypeInfo.getPermitRunEndTime();
                    CronExpression cexpEnd = new CronExpression(tmpEndStr);
                    Date firstEndTime = cexpEnd.getNextValidTimeAfter(firstStartTime);
                    Date nowEndTime = cexpEnd.getNextValidTimeAfter(current);
                    if (!nowEndTime.equals(firstEndTime) && current.before(nowEndTime)) {
                        isRunNow = true;
                        firstEndTime = nowEndTime;
                    }
                    this.heartBeatTimer.schedule(
                            new PauseOrResumeTask(this, this.heartBeatTimer,
                                    PauseOrResumeTask.TYPE_PAUSE, tmpEndStr),
                            firstEndTime);
                    this.currenServer.setNextRunEndTime(ScheduleUtil.transferDataToString(firstEndTime));
                } catch (Exception e) {
                    logger.error("�����һ��ִ��ʱ������쳣:" + currenServer.getUuid(), e);
                    throw new Exception("�����һ��ִ��ʱ������쳣:" + currenServer.getUuid(), e);
                }
            }
        }
        if (isRunNow == true) {
            this.resume("����������������");
        }
        this.rewriteScheduleInfo();

    }

    /**
     * ��Processû�л�ȡ�����ݵ�ʱ����ã������Ƿ���ʱֹͣ������
     *
     * @throws Exception
     */
    public boolean isContinueWhenData() throws Exception {
        if (isPauseWhenNoData() == true) {
            this.pause("û������,��ͣ����");
            return false;
        } else {
            return true;
        }
    }

    public boolean isPauseWhenNoData() {
        //�����û�з��䵽������������˳�
        if (this.currentTaskItemList.size() > 0 && this.taskTypeInfo.getPermitRunStartTime() != null) {
            if (this.taskTypeInfo.getPermitRunEndTime() == null
                    || this.taskTypeInfo.getPermitRunEndTime().equals("-1")) {
                return true;
            } else {
                return false;
            }
        } else {
            return false;
        }
    }

    /**
     * �������е�����ʱ�䣬��ʱֹͣ����
     *
     * @throws Exception
     */
    public void pause(String message) throws Exception {
        if (this.isPauseSchedule == false) {
            this.isPauseSchedule = true;
            this.pauseMessage = message;
            if (logger.isDebugEnabled()) {
                logger.debug("��ͣ���� ��" + this.currenServer.getUuid() + ":" + this.statistics.getDealDescription());
            }
            if (this.processor != null) {
                this.processor.stopSchedule();
            }
            rewriteScheduleInfo();
        }
    }

    /**
     * �����˿�ִ�е�ʱ�����䣬�ָ�����
     *
     * @throws Exception
     */
    public void resume(String message) throws Exception {
        if (this.isPauseSchedule == true) {
            if (logger.isDebugEnabled()) {
                logger.debug("�ָ�����:" + this.currenServer.getUuid());
            }
            this.isPauseSchedule = false;
            this.pauseMessage = message;
            if (this.task != null) {
                if (this.taskTypeInfo.getProcessorType() != null &&
                        this.taskTypeInfo.getProcessorType().equalsIgnoreCase("NOTSLEEP") == true) {
                    this.taskTypeInfo.setProcessorType("NOTSLEEP");
                    this.processor = new ProcessorNotSleep(this,
                            task, this.statistics);
                } else {
                    this.processor = new ProcessorSleep(this,
                            task, this.statistics);
                    this.taskTypeInfo.setProcessorType("SLEEP");
                }
            }
            rewriteScheduleInfo();
        }
    }

    /**
     * ��������ֹͣ��ʱ�򣬵��ô˷����������δ�������������������ע����Ϣ��
     * Ҳ�����ǿ������ķ������ָֹ�
     * ��Ҫע����ǣ�������������ڵ�ǰ��������Ϻ����ִ��
     *
     * @throws Exception
     */
    public void stop(String strategyName) throws Exception {
        if (logger.isInfoEnabled()) {
            logger.info("ֹͣ������ ��" + this.currenServer.getUuid());
        }
        this.isPauseSchedule = false;
        if (this.processor != null) {
            this.processor.stopSchedule();
        } else {
            this.unRegisterScheduleServer();
        }
    }

    /**
     * ֻӦ����Processor�е���
     *
     * @throws Exception
     */
    public void unRegisterScheduleServer() throws Exception {
        registerLock.lock();
        try {
            if (this.processor != null) {
                this.processor = null;
            }
            if (this.isPauseSchedule == true) {
                // ����ͣ���ȣ���ע��Manager�Լ�
                return;
            }
            if (logger.isDebugEnabled()) {
                logger.debug("ע�������� ��" + this.currenServer.getUuid());
            }
            this.isStopSchedule = true;
            // ȡ������TIMER
            this.heartBeatTimer.cancel();
            // ����������ע���Լ�
            this.scheduleCenter.unRegisterScheduleServer(
                    this.currenServer.getTaskType(),
                    this.currenServer.getUuid());
        } finally {
            registerLock.unlock();
        }
    }

    public TaskType getTaskTypeInfo() {
        return taskTypeInfo;
    }


    public Statistics getStatistics() {
        return statistics;
    }

    /**
     * ��ӡ�����������͵�����������
     *
     * @param taskType
     */
    public void printScheduleServerInfo(String taskType) {

    }

    public Server getScheduleServer() {
        return this.currenServer;
    }

    public String getmBeanName() {
        return mBeanName;
    }
}

